/*
 * Copyright 2023 AntGroup CO., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 */

package com.antgroup.geaflow.shuffle.serialize;

import com.antgroup.geaflow.common.serialize.SerializerFactory;
import com.antgroup.geaflow.common.serialize.impl.KryoSerializer;
import com.antgroup.geaflow.shuffle.api.pipeline.buffer.OutBuffer.BufferBuilder;
import com.esotericsoftware.kryo.io.Output;

public class RecordSerializer<T> implements IRecordSerializer<T> {

    private final Output output;
    private final KryoSerializer kryoSerializer;

    public RecordSerializer() {
        output = new Output(4096);
        kryoSerializer = ((KryoSerializer) SerializerFactory.getKryoSerializer());
    }

    @Override
    public void serialize(T value, boolean isRetract, BufferBuilder outBuffer) {
        Output output = this.output;
        output.setOutputStream(outBuffer.getOutputStream());
        try {
            kryoSerializer.getThreadKryo().writeClassAndObject(output, value);
            output.flush();
        } finally {
            output.clear();
        }
    }

}
